package cn.regexp.elasticjob.service;

import cn.regexp.elasticjob.config.Job;
import cn.regexp.elasticjob.controller.JobController;
import cn.regexp.elasticjob.domain.SysJob;
import cn.regexp.elasticjob.mapper.JobMapper;
import cn.regexp.elasticjob.util.JsonUtils;
import com.dangdang.ddframe.job.api.ElasticJob;
import com.dangdang.ddframe.job.api.script.ScriptJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.exception.JobConfigurationException;
import com.dangdang.ddframe.job.exception.JobSystemException;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.internal.schedule.*;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import com.google.common.base.Optional;
import org.apache.curator.CuratorZookeeperClient;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.api.GetDataBuilder;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.shardingsphere.elasticjob.infra.yaml.YamlEngine;
import org.apache.shardingsphere.elasticjob.lite.lifecycle.api.JobOperateAPI;
import org.apache.zookeeper.ZooKeeper;
import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.impl.StdSchedulerFactory;
import org.quartz.simpl.SimpleThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Objects;
import java.util.Properties;

@Service
public class JobService {

    private Logger logger = LoggerFactory.getLogger(JobService.class);

    @Autowired
    private ZookeeperRegistryCenter zookeeperRegistryCenter;

    @Autowired
    private ApplicationContext ctx;
    @Autowired
    private JobMapper jobMapper;

    /**
     * 封装Job数据
     * @param sysJob
     * @return
     */
    public Job pakageJob(SysJob sysJob) {
        //基本数据封装
        Job job = new Job();
        job.setJobName(sysJob.getJobName());
        job.setCron(sysJob.getCron());
        job.setJobType(sysJob.getJobType());
        job.setShardingTotalCount(sysJob.getShardTotalCount());
        job.setJobParameter(sysJob.getShardItemParam());
        job.setJobClass(sysJob.getJobClass());
        //脚本文件需单独处理
        if ( "SCRIPT".equals(sysJob.getJobType())){
            String scriptFilePath = sysJob.getScriptFilePath();
            String scriptLines = sysJob.getScriptLines();
            String scriptType = sysJob.getScriptType();
            //命令行和脚本文件同时存在时优先考虑脚本文件
            if (StringUtils.hasText(scriptFilePath)){
                if (StringUtils.hasText(scriptType)) {
                    job.setScriptCommandLine(scriptType + " " + scriptFilePath);
                }
            }else if (StringUtils.hasText(scriptLines)){
                if (StringUtils.hasText(scriptType)) {
                    job.setScriptCommandLine(scriptType + " " + scriptLines);
                }
            }
        }
        return job;
    }

    /**
     * 封装定时任务的配置
     * @param job
     * @return
     */
    public LiteJobConfiguration pakegeConfiguration(Job job) {
        // 核心配置
        JobCoreConfiguration coreConfig =
                JobCoreConfiguration.newBuilder(job.getJobName(), job.getCron(), job.getShardingTotalCount())
                        .shardingItemParameters(job.getShardingItemParameters())
                        .description(job.getDescription())
                        .failover(job.isFailover())
                        .jobParameter(job.getJobParameter())
                        .misfire(job.isMisfire())
                        //            .jobProperties(JobPropertiesEnum.JOB_EXCEPTION_HANDLER.getKey(), job.getJobProperties().getJobExceptionHandler())
                        //             .jobProperties(JobPropertiesEnum.EXECUTOR_SERVICE_HANDLER.getKey(), job.getJobProperties().getExecutorServiceHandler())
                        .build();

        // 不同类型的任务配置处理
        JobTypeConfiguration typeConfig = null;
        String jobType = job.getJobType();
        if (jobType.equals("SIMPLE")) {
            typeConfig = new SimpleJobConfiguration(coreConfig, job.getJobClass());
        }

        if (jobType.equals("DATAFLOW")) {
            typeConfig = new DataflowJobConfiguration(coreConfig, job.getJobClass(), job.isStreamingProcess());
        }

        if (jobType.equals("SCRIPT")) {
            typeConfig = new ScriptJobConfiguration(coreConfig, job.getScriptCommandLine());
        }

       return LiteJobConfiguration.newBuilder(typeConfig)
                .overwrite(job.isOverwrite())
                .disabled(job.isDisabled())
                .monitorPort(job.getMonitorPort())
                .monitorExecution(job.isMonitorExecution())
                .maxTimeDiffSeconds(job.getMaxTimeDiffSeconds())
                .jobShardingStrategyClass(job.getJobShardingStrategyClass())
                .reconcileIntervalMinutes(job.getReconcileIntervalMinutes())
                .build();
    }

    /**
     * 注册定时任务
     * @param job
     * @param startNow 是否立即启动
     */
    public void registerJob(Job job,Boolean startNow) {
        String jobName = job.getJobName();
        //封装定时任务配置
        LiteJobConfiguration jobConfig = pakegeConfiguration(job);

        List<BeanDefinition> elasticJobListeners = getTargetElasticJobListeners(job);

        // 构建SpringJobScheduler对象来初始化任务
        BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(SpringJobScheduler.class);
        factory.setScope(BeanDefinition.SCOPE_PROTOTYPE);
        if ("SCRIPT".equals(job.getJobType())) {
            factory.addConstructorArgValue(null);
        } else {
            BeanDefinitionBuilder rdbFactory = BeanDefinitionBuilder.rootBeanDefinition(job.getJobClass());
            factory.addConstructorArgValue(rdbFactory.getBeanDefinition());
        }
        factory.addConstructorArgValue(zookeeperRegistryCenter);
        factory.addConstructorArgValue(jobConfig);

        // 任务执行日志数据源，以名称获取
        if (StringUtils.hasText(job.getEventTraceRdbDataSource())) {
            BeanDefinitionBuilder rdbFactory = BeanDefinitionBuilder.rootBeanDefinition(JobEventRdbConfiguration.class);
            rdbFactory.addConstructorArgReference(job.getEventTraceRdbDataSource());
            factory.addConstructorArgValue(rdbFactory.getBeanDefinition());
        }
        //将启动的动态定时任务加入SpringJobScheduler中
        factory.addConstructorArgValue(elasticJobListeners);
        DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) ctx.getAutowireCapableBeanFactory();

        if(!defaultListableBeanFactory.containsBeanDefinition("SpringJobScheduler" + jobName)){
            defaultListableBeanFactory.registerBeanDefinition("SpringJobScheduler" + jobName, factory.getBeanDefinition());
        }

        SpringJobScheduler springJobScheduler = (SpringJobScheduler) ctx.getBean("SpringJobScheduler" + job.getJobName());
        if (startNow){
            springJobScheduler.init();
            logger.info("【" + job.getJobName() + "】\t" + job.getJobClass() + "\tinit success");
        }

    }


    /**
     * 获取定时任务的错误日志配置
     * @param job
     * @return
     */
    private List<BeanDefinition> getTargetElasticJobListeners(Job job) {
        List<BeanDefinition> result = new ManagedList<BeanDefinition>(2);
        String listeners = job.getListener();
        if (StringUtils.hasText(listeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(listeners);
            factory.setScope(BeanDefinition.SCOPE_PROTOTYPE);
            result.add(factory.getBeanDefinition());
        }

        String distributedListeners = job.getDistributedListener();
        long startedTimeoutMilliseconds = job.getStartedTimeoutMilliseconds();
        long completedTimeoutMilliseconds = job.getCompletedTimeoutMilliseconds();

        if (StringUtils.hasText(distributedListeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(distributedListeners);
            factory.setScope(BeanDefinition.SCOPE_PROTOTYPE);
            factory.addConstructorArgValue(startedTimeoutMilliseconds);
            factory.addConstructorArgValue(completedTimeoutMilliseconds);
            result.add(factory.getBeanDefinition());
        }
        return result;
    }

    /**
     * 将定时任务停止并删除
     * @param jobName
     */
    public void removeJob(String jobName) {
        JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobName);
        if (jobScheduleController != null) {
            //暂停任务
            jobScheduleController.pauseJob();
            //关闭调度器
            jobScheduleController.shutdown();
            //删除节点
            zookeeperRegistryCenter.remove("/" + jobName);
            /**
             * 删除存储在application中的Bean
             */
             //1.将applicationContext转化为ConfigurableApplicationContext
            ConfigurableApplicationContext context = (ConfigurableApplicationContext) ctx;
            //2.获取bean工厂并转换为DefaultListableBeanFactory
            DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) context.getBeanFactory();
            // 动态删除
            beanFactory.removeBeanDefinition("SpringJobScheduler" + jobName);
        }

    }

    /**
     * 暂停定时任务
     * @param jobName
     */
    public void stopJob(String jobName) {
        JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobName);
        if (jobScheduleController != null) {
            //暂停任务
            jobScheduleController.pauseJob();
        }

    }
    /**
     * 触发定时任务（执行一次）
     * @param jobName
     */
    public void triggerJob(String jobName) {
        JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobName);
        if (jobScheduleController != null) {
            //触发任务
            jobScheduleController.triggerJob();
        }
    }
    /**
     * 启动定时任务
     * @param jobName
     */
    public void startJob(String jobName) {
        if (ctx.getBean("SpringJobScheduler" + jobName) != null) {
            //触发任务
            SpringJobScheduler springJobScheduler = (SpringJobScheduler) ctx.getBean("SpringJobScheduler" + jobName);
            springJobScheduler.init();
        }else {
            try {
                throw new Exception("定时任务还未注册请勿直接启动！");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }
    /**
     * 恢复被暂停的定时任务
     * @param jobName
     */
    public void recoverJob(String jobName) {
        JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobName);

        if (jobScheduleController != null) {
            //执行定时任务
            jobScheduleController.resumeJob();
        }
    }



    public void getJobs() throws Exception {
        CuratorFramework client = zookeeperRegistryCenter.getClient();
        CuratorZookeeperClient zookeeperClient = client.getZookeeperClient();
        ZooKeeper zooKeeper = zookeeperClient.getZooKeeper();
        List<String> childrens = zooKeeper.getChildren("/", false);
        for (String children : childrens) {
            GetChildrenBuilder children1 = client.getChildren();

            GetDataBuilder data = client.getData();
            byte[] bytes = data.forPath("/"+children+"/config");
            String string =new String(bytes);
        }
    }


    /**
     * 获取zookeeper上的定时任务配置
     * @param jobName
     * @return
     * @throws Exception
     */
    public Job getConfig(String jobName) throws Exception {
        //若定时任务已注册
        if (JobRegistry.getInstance().getJobScheduleController(jobName) != null) {
            CuratorFramework client = zookeeperRegistryCenter.getClient();
            GetDataBuilder data = client.getData();
            String config = new String(data.forPath("/" + jobName + "/config"));
            return JsonUtils.toBean(Job.class, config);
        }
        return pakageJob(jobMapper.getByJobName(jobName));
    }

    /**
     * 更新定时任务的配置
     * @param job
     * @throws Exception
     */
    public void updateConfig(Job job)  {
        // 核心配置
        LiteJobConfiguration jobConfig = pakegeConfiguration(job);
        //若该任务已启动 更新服务器配置
        if (JobRegistry.getInstance().getJobScheduleController(job.getJobName()) != null) {
            SpringJobScheduler jobScheduler = (SpringJobScheduler) ctx.getBean("SpringJobScheduler" + jobConfig.getJobName());
            SchedulerFacade schedulerFacade = jobScheduler.getSchedulerFacade();
            schedulerFacade.updateJobConfiguration(jobConfig);
            jobScheduler.init();
            String configPath = "/" + job.getJobName() + "/config";
            String s = JsonUtils.toJson(job);
            zookeeperRegistryCenter.update(configPath, s);
        }
        //更新数据库中配置
        updateSysJob(job);
    }

    private int updateSysJob(Job job){
        SysJob sysJob = jobMapper.getByJobName(job.getJobName());
        BeanUtils.copyProperties(job, sysJob);
        return jobMapper.update(sysJob);
    }

    /**
     * 开启任务监听,当有任务添加时，监听zk中的数据增加，自动在其他节点也初始化该任务
     */
    public void monitorJobRegister() {
        CuratorFramework client = zookeeperRegistryCenter.getClient();
        @SuppressWarnings("resource")
        PathChildrenCache childrenCache = new PathChildrenCache(client, "/", true);
        PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                ChildData data = event.getData();
                switch (event.getType()) {
                    case CHILD_ADDED:
                        String config = new String(client.getData().forPath(data.getPath() + "/config"));
                        Job job = JsonUtils.toBean(Job.class, config);
                        Object bean = null;
                        // 获取bean失败则添加任务
                        try {
                            bean = ctx.getBean("SpringJobScheduler" + job.getJobName());
                        } catch (BeansException e) {
                            logger.error("ERROR NO BEAN,CREATE BEAN SpringJobScheduler" + job.getJobName());
                        }
                        if (Objects.isNull(bean)) {
                            registerJob(job,false);
                        }
                        break;
                    default:
                        break;
                }
            }
        };
        childrenCache.getListenable().addListener(childrenCacheListener);
        try {
            childrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            //childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
